package gdingtalk

import (
	"fmt"
	"github.com/astaxie/beego/logs"
	"github.com/tidwall/gjson"
	"gorm.io/gorm/clause"
	"time"
)

type DingCloudPush struct {
	subscribeId string
	bizType     []int64
	event       map[int]func(o OpenSyncBizData) error
}

// NewDingCloudPush 初始化
func NewDingCloudPush(loopBizType []int64) (DingCloudPush, error) {
	return DingCloudPush{
		subscribeId: fmt.Sprintf("%d_0", gjson.Get(gOption.DingParams, "suiteId").Int()),
		bizType:     loopBizType,
		event:       make(map[int]func(o OpenSyncBizData) error),
	}, nil
}

// Start 循环处理消息
func (d *DingCloudPush) Start() {
	if gOption.Type != "isv" {
		logs.Warn("type %s is not support", gOption.Type)
		return
	}
	for {
		msgs, err := d.getSyncData(10)
		if err != nil {
			logs.Error("getSyncData err:%s", err)
		}
		// 没有数据暂停500毫秒
		if len(msgs) <= 0 {
			time.Sleep(time.Millisecond * 500)
			continue
		}
		for _, v := range msgs {
			d.doMsg(v)
		}
		// 处理成功后休息50ms，防止处理太快，造成数据库压力过大
		time.Sleep(time.Millisecond * 50)
	}
}

// SubscribeEvent 注册事件
func (d *DingCloudPush) SubscribeEvent(bizType int, cb func(o OpenSyncBizData) error) {
	d.event[bizType] = cb
}

// doMsg 处理消息
func (d *DingCloudPush) doMsg(o OpenSyncBizData) {
	if v, ok := d.event[int(o.BizType)]; ok {
		if err := v(o); err == nil {
			o.UpdateStatus(StatusSuccess)
		} else {
			o.UpdateStatus(StatusFail)
			logs.Error("ding_cloud_push id:%d 处理失败, err:%s", o.ID, err)
			// todo: 报警
		}
	} else {
		o.UpdateStatus(StatusFail)
		logs.Error("biz_type:%d 没有找到处理程序", o.BizType)
		// todo: 报警
	}
}

func (d *DingCloudPush) getSyncData(count int) ([]OpenSyncBizData, error) {
	var bizs []OpenSyncBizData
	bizs, err := d.getSync(count, "open_sync_biz_data")
	if err != nil {
		return []OpenSyncBizData{}, err
	}
	// 如果不足count条，从medium中补足
	medium := count - len(bizs)
	if medium <= 0 {
		return bizs, nil
	}
	var bizMediums []OpenSyncBizData
	bizMediums, err = d.getSync(count, "open_sync_biz_data_medium")
	if err != nil {
		return bizs, err
	}
	return append(bizs, bizMediums...), nil
}

func (d *DingCloudPush) getSync(count int, tableName string) ([]OpenSyncBizData, error) {
	var bizs []OpenSyncBizData
	tx := db.Begin()
	res := tx.Table(tableName).Where("subscribe_id = ?", d.subscribeId).
		Where("status = ?", StatusWait).Where("biz_type in ?", d.bizType).
		Clauses(clause.Locking{Strength: "UPDATE"}).Limit(count).Find(&bizs)
	if res.Error != nil {
		tx.Commit()
		return []OpenSyncBizData{}, res.Error
	}
	if len(bizs) <= 0 {
		tx.Commit()
		return []OpenSyncBizData{}, nil
	}
	// 把状态改成处理中，不用等待处理完再释放事务，减少冲突
	var ids []int64
	for _, v := range bizs {
		ids = append(ids, v.ID)
	}
	res = tx.Table(tableName).Where("id in ?", ids).Update("status", StatusDoing)
	if res.Error != nil {
		tx.Rollback()
		return []OpenSyncBizData{}, res.Error
	}
	d.setTableName(bizs, tableName)
	tx.Commit()
	return bizs, nil
}

// setTableName替换表名
// tableName, 可选值open_sync_biz_data， open_sync_biz_data_medium
func (d *DingCloudPush) setTableName(bizs []OpenSyncBizData, tableName string) {
	for k, _ := range bizs {
		bizs[k].TableName = tableName
	}
}
